Beam attenuation#35
Conversation
…ng tests; ignored eiger_async tests for now so that running tests is somewhat useful
Jakub Wlodek (jwlodek)
left a comment
Jakub Wlodek (jwlodek)
left a comment
There was a problem hiding this comment.
Marked some suggestions. Mainly I'd recommend making both device classes Movables and re-work the current methods for changing states into a set method override.
| return np.exp(-self.linear_atten_coefficient * self.thickness_cm) | ||
|
|
||
|
|
||
| class AttenuatorBank(StandardReadable, EpicsDevice): |
There was a problem hiding this comment.
| class AttenuatorBank(StandardReadable, EpicsDevice): | |
| class AttenuatorBank(StandardReadable, EpicsDevice, Movable[float]): |
| *(a.get_status() for _, a in self.attenuators.items()) | ||
| ) | ||
|
|
||
| async def set_attenuation(self, target_attenuation: float): |
There was a problem hiding this comment.
| async def set_attenuation(self, target_attenuation: float): | |
| from ophyd_async.core import AsyncStatus | |
| @AsyncStatus.wrap | |
| async def set(self, target_attenuation: float): |
I suggest making this top level device a Movable. This way, in a plan, you can do:
yield from bps.mv(attenuator_bank, 20.0)or directly via the RunEngine:
RE(bps.mv(attenuator_bank. 20.0))instead of calling set_attenuation directly. The benefit if this is that you can do things like pause and wait at this step if the beam goes down, you can "dry-run" a plan without actually moving anything, and generally you get all the tooling around plans this way. It also allows you to call this and either have it block and wait for it to be done, or add it to a group of other moves and wait for them all in parallel.
| class Attenuator(EpicsDevice): | ||
| filter_material = "Al" | ||
| filter_material_z = 13 | ||
| filter_density = xraylib.ElementDensity(filter_material_z) # g/cm³ |
There was a problem hiding this comment.
Are these always these values, or can they be changed?
There was a problem hiding this comment.
They are static for now and cannot change for the existing filters. Garth plans to get more filters and I don't know if they will be aluminum or something else.
| HIGH = "High" # on | ||
|
|
||
|
|
||
| class Attenuator(EpicsDevice): |
There was a problem hiding this comment.
| class Attenuator(EpicsDevice): | |
| class Attenuator(EpicsDevice, Movable[AttenuatorEnum]): |
Similar to the higher level class, I'd recommend making this a Movable.
There was a problem hiding this comment.
For ophyd-async, this would be an AsyncMovable right?
There was a problem hiding this comment.
Nope, for protocols there's no Sync vs Async: https://github.com/bluesky/bluesky/blob/4418a6b16cb3da6626f2b07e8054de3fd28591e5/src/bluesky/protocols.py#L334
| """ | ||
| self.prefix = prefix | ||
| self.num = num | ||
| self.cmd = epics_signal_w(AttenuatorEnum, f"{self.prefix}:DO{self.num + 1}-Cmd") |
There was a problem hiding this comment.
Is the in_status signal the one that represents the actual physical state of the attenuator? If yes, I'd recommend consolidating these into one epics_signal_rw, where you set the read and write pvs to different values. Then, the set method can just call await self.cmd.set(value). In this scenario, maybe I'd rename the signal to state or position.
In this case I assume that the open/close operations are pretty quick, but still, generally we want to make sure we reach our target before continuing, unless the user explicitly says not to. Basically, We want:
yield from bps.mv(attenuator, AttenuatorEnum.LOW)to block by default until we have confirmation that the attenuator has reached its target position.
…syncStatus, upgraded version of ophyd-async because of test methods used, and to match what is in cdi-profile-collection
…d-async is incompatible with it
2 participants
No description provided.